package com.chenjl.api;

import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerConfig;
/**
 * 消费者配置
 * new API offset不在client 维护，由broker在zk维护
 * 
 * 2016-9-14 11:02:15
 * @author chenjinlong
 */
public class TestOnConsumerConfig {
	public static void main(String[] args) {
		Properties props = new Properties();
		
		//标记client端的一个编码值, 同一个进程内，多个client端时，如果没有指定，默认根据1这个值向后增加
		props.put("client.id","1001A");
		props.put("group.id","group01");
		props.put("bootstrap.servers","192.168.245.129:9092,192.168.245.130:9092");
		props.put("key.deserializer","org.apache.kafka.common.serialization.StringSerializer");
		props.put("value.deserializer","org.apache.kafka.common.serialization.StringSerializer");
		//
		props.put("max.poll.records",20);
		//
		props.put("session.timeout.ms",30000);
		props.put("heartbeat.interval.ms",3000);
		//
		props.put("enable.auto.commit",true);
		props.put("auto.commit.interval.ms",5000);
		
		//("latest", "earliest", "none"),
		props.put("auto.offset.reset","latest");
		//
		props.put("fetch.min.bytes",1);
		props.put("fetch.max.wait.ms",500);
		//配置每个producer端缓存topic的metadata的过期时间，默认值为5分钟
		props.put("metadata.max.age.ms",5*60*1000);
		//
		props.put("max.partition.fetch.bytes",1*1024*1024);		
		//用于设置socket的发送缓冲区SO_SNDBUF的大小
		props.put("send.buffer.bytes",128 * 1024);
		//用于设置socket的接收响应的缓冲区SO_RCVBUF的大小
		props.put("receive.buffer.bytes",64 * 1024);
		//用于设置重新尝试连接的等待时间
		props.put("reconnect.backoff.ms",50);
		//在向broker发送数据失败后的重试间隔时间，默认值为100ms
		props.put("retry.backoff.ms",100L);		
		
		//
		props.put("check.crcs",true);
		//设置连接最大的空闲时间
		props.put("connections.max.idle.ms",9*60*1000);
		//配置socket请求的最大超时时间，默认值为30秒
		props.put("request.timeout.ms",30*1000);
		//props.put("interceptor.classes","");
		//props.put("exclude.internal.topics","");
		//props.put("partition.assignment.strategy","");
		
		props.put("security.protocol","PLAINTEXT");
		
		ConsumerConfig consumerConfig = new ConsumerConfig(props);
		System.out.println(consumerConfig.toString());
	}
}